/*
 * @(#)BinlogEventSource.java   1.0  2021年11月8日
 * 
 * Copyright (c)	2014-2020. All Rights Reserved.	GuangZhou hhmk Technology Company LTD.
 */
package com.swift.dts.source;

import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import com.alibaba.otter.canal.protocol.ClientIdentity;
import com.alibaba.otter.canal.protocol.Message;
import com.swift.dts.source.canal.CanalDtsServer;
import com.swift.dts.source.thread.BinlogTableThreadExec;
import com.swift.util.exec.ThreadUtil;
import com.swift.util.type.JsonUtil;

/**
 * binlog事件来源接收器
 * 
 * @author zhengjiajin
 * @version 1.0 2021年11月8日
 */
public class BinlogEventSource extends Thread {

    private static Logger log = LoggerFactory.getLogger(BinlogEventSource.class);
    
    private ClientIdentity clientIdentity;

    private boolean stop = false;
    
    @Autowired
    private CanalDtsServer canalDtsServer;
    
    @Autowired
    private BinlogTableThreadExec binlogTableThreadExec;
    
    public BinlogEventSource(ClientIdentity clientIdentity) {
        this.setDaemon(true);
        this.clientIdentity = clientIdentity;
    }
    
    public void run() {
        // 需要cache
        while (!stop) {
            try {
                Message message = canalDtsServer.getWithoutAck(clientIdentity, 10, 1l, TimeUnit.SECONDS);
                if (message.getId() != -1) {
                    binlogTableThreadExec.put(clientIdentity, message);
                } else {
                    log.info(clientIdentity.getDestination()+" Message id is -1:" + JsonUtil.toJson(message));
                    ThreadUtil.sleep(1000);
                }
            } catch (Exception ex) {
                log.error("接收消息异常:", ex);
                ThreadUtil.sleep(1000);
            }
        }

    }


    /**
     * @return the stop
     */
    public boolean isStop() {
        return stop;
    }


    /**
     * @param stop the stop to set
     */
    public void setStop(boolean stop) {
        this.stop = stop;
    }


    /**
     * @return the clientIdentity
     */
    public ClientIdentity getClientIdentity() {
        return clientIdentity;
    }


    /**
     * @param clientIdentity the clientIdentity to set
     */
    public void setClientIdentity(ClientIdentity clientIdentity) {
        this.clientIdentity = clientIdentity;
    }
    
}
